package parallels.servertech.server.nio;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.logging.Level;
import java.util.logging.Logger;

import parallels.servertech.common.Status;

public class Decoder implements Runnable {
	private static final Logger log = Logger.getLogger("decoder");
	
	private static final int SELECTION_TIMEOUT = 1000;
	
	private final int port;
	private final Thread decodingThread;
	
	private volatile boolean stopped = false;
	
	public Decoder(int port) {
		this.port = port;
		decodingThread = new Thread(this, "decoder");
	}
	
	private static class InputOutputPair {
		private final DataInputStream in;
		private final DataOutputStream out;
		
		public InputOutputPair(SocketChannel channel) {
			this.in = new DataInputStream(new NonBlockingChannelInputStream(channel));
			this.out = new DataOutputStream(new NonBlockingChannelOutputStream(channel));
		}
	}

	@Override
	public void run() {
		try {
			ServerSocketChannel serverChannel = ServerSocketChannel.open();
			serverChannel.bind(new InetSocketAddress(port));
			serverChannel.configureBlocking(false);
			
			Selector selector = Selector.open();
			serverChannel.register(selector, SelectionKey.OP_ACCEPT);
			
			while (!stopped) {
				int ready = selector.select(SELECTION_TIMEOUT);
				if (ready > 0) {
					log.info(ready + " channels are ready");
					for (Iterator<SelectionKey> keysIter = selector.selectedKeys().iterator();keysIter.hasNext();) {
						SelectionKey key = keysIter.next();
						
						if (key.isAcceptable()) {
							SocketChannel channel = ((ServerSocketChannel)key.channel()).accept();
							log.info("client " + channel.getRemoteAddress() + " connected");
							
							channel.configureBlocking(false);
							InputOutputPair inOut = new InputOutputPair(channel);
							// TODO ability to write will be detected when command output will be ready
							channel.register(selector, SelectionKey.OP_READ, inOut);
						} else if (key.isReadable()) {
							SocketChannel channel = (SocketChannel) key.channel();
							log.info("client channel @" + channel.hashCode() + " ready: " + key.readyOps());
							decode((InputOutputPair)key.attachment());
						}
						
						keysIter.remove();
					}
				}
			}
			
			closeChannels(selector);
			selector.close();
			serverChannel.close();
			
		} catch (IOException ioe) {
			log.log(Level.SEVERE, "demultiplexing error: " + ioe.getClass().getName() + "(" + ioe.getMessage() + ")");
		}
	}

	private void closeChannels(Selector selector) {
		for (SelectionKey key : selector.keys()) {
			try {
				key.channel().close();
			} catch (IOException closeEx) {
				log.warning("failed to close channel");
			}
		}
	}

	private void decode(InputOutputPair inOut) {
		log.info("decoding command");
		// TODO move command execution to dedicated pool
		try {
			String cmd = inOut.in.readUTF();
			if ("add".equals(cmd)) {
				int left = inOut.in.readInt();
				int right = inOut.in.readInt();
				log.info("adding " + left + " and " + right);
				inOut.out.writeInt(Status.SUCCESS.ordinal());
				inOut.out.writeInt(left + right);
			} else if ("shutdown".equals(cmd)) {
				inOut.out.writeInt(Status.SUCCESS.ordinal());
				log.info("shutting down...");
				stopped = true;
			} else {
				inOut.out.writeInt(Status.FAILURE.ordinal());
				inOut.out.writeUTF("unknown command: '" + cmd + "'");
			}
		} catch (IOException ioe) {
			log.log(Level.SEVERE, "command decoding error: " + ioe.getClass().getName() + "(" + ioe.getMessage() + ")");
		}
	}

	public void start() {
		log.info("starting decoder");
		decodingThread.start();
		log.info("decoder started");
	}
}
